package hasthi.manager;

import static hasthi.common.constants.HasthiConstants.FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY;
import hasthi.common.BaseContext;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.LogFile;
import hasthi.common.RemoteProcessorException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.HasthiConstants.MngState;
import hasthi.common.constants.WSDMConstants.SystemState;
import hasthi.container.CommunicationClient;
import hasthi.container.MessageContext;
import hasthi.container.NodeAddress;
import hasthi.container.SimpleMessageContext;
import hasthi.instrument.FailureDetector;
import hasthi.instrument.JavaInstrumentation;
import hasthi.manager.GetCoordinatorResponseDocument.GetCoordinatorResponse;
import hasthi.manager.ManagerHeartBeatResponseDocument.ManagerHeartBeatResponse;
import hasthi.manager.ManagerHeartbeatDocument.ManagerHeartbeat;
import hasthi.manager.ManagerPingResponseDocument.ManagerPingResponse;
import hasthi.manager.monitering.VerificationTools;
import hasthi.metamodel.ChangeBuffer;
import hasthi.metamodel.ChangeEvent;
import hasthi.metamodel.MixedResourceModel;
import hasthi.metamodel.ResourceUtils;
import hasthi.metamodel.ChangeEvent.ChangeType;
import hasthi.metamodel.resource.EditableManagedResource;
import hasthi.metamodel.resource.ManagedResource;
import hasthi.tools.ManagementRingTools;
import hasthi.tools.analyze.DataCollector;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class ManagementDemon implements Runnable {
    private ManagerContext managerContext;
    private List<QName> resourcesToRemove = new ArrayList<QName>();
    /**
     * We use this map to build the heartbeat event. Since only one thread access this
     * there should not be any race conditions
     */
    
    private DistributedLogger dLogger;
    
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    
    private long coordinatorTimestamp = Long.MAX_VALUE;
    private ManagerAddress coodrinatorAddress;
    private DataCollector loc;
    private DataCollector roc;
    private DataCollector moc;
    
    private ScheduledFuture demonHandle;

    public ManagementDemon(ManagerContext managerContext) {
        this.managerContext = managerContext;
        dLogger = managerContext.getBaseContext().getDlogger();
        //This code collect statistics about the management system
        LogFile logfile = LogFile.getLog("hasthi_manager");
        loc = new DataCollector("loc(mng"+UUID.randomUUID()+")",logfile,DataCollector.WINDOW_SIZE);
        roc = new DataCollector("roc(mng"+UUID.randomUUID()+")",logfile,DataCollector.WINDOW_SIZE);
        moc = new DataCollector("moc(mng"+UUID.randomUUID()+")",logfile,DataCollector.WINDOW_SIZE);
    }
    
    public void startDeamon() throws HasthiException{
        if(demonHandle != null){
            throw new HasthiException("Demon Already started",FaultCode.LocalError);
        }
        //Timer timer = new Timer(true);
        int delay = 5000; // delay for 5 sec.
        int period = managerContext.getBaseContext().getManagerEpochSeconds()*1000;
        log.info("Using " + period + "ms as manager epoch time");
        demonHandle = managerContext.getScheduledExecutorService().scheduleAtFixedRate(this, 
                delay, period, TimeUnit.MILLISECONDS);
    }
    
    public void stopDeamon() throws HasthiException{
        if(demonHandle == null){
            throw new HasthiException("Demon Not started",FaultCode.LocalError);
        }
        demonHandle.cancel(true);
    }
    

    public void run() {
        long point1 = System.nanoTime();
        long point12 = -1;
        long point13 = -1;
        try {
            BaseContext baseContext = managerContext.getBaseContext();
            
            ManagerAddress address = managerContext.getMyAddress();
            long failedResourceCount = 0;
            int id;
            if(address != null){
                id = address.hashCode();
            }else{
                id = 0;
            }
            
            //System.out.println(msg);
            long currentTime = System.currentTimeMillis();
            /**
             * Check the last updated time of each resource, mark the those
             * whose lease has expired as down
             */
            failedResourceCount = 0;
            MixedResourceModel resourceModel = managerContext.getResourceModel();
            
            final String stoppedStateStr = SystemState.StoppedState.toString();
            long managerEpoch2 = 2 * baseContext.getMngAgentEpochSeconds()* 1000;
            long managerEpoch5 = 5 * baseContext.getMngAgentEpochSeconds()* 1000;
            
            Collection<QName> names = resourceModel.getResources();
            for (QName name : names) {
                EditableManagedResource resource = (EditableManagedResource) resourceModel.getResource(name);
                long timeSinceLastUpdate = currentTime - resource.getLastUpdated(); 
                
                if(timeSinceLastUpdate > managerEpoch2){
                    failedResourceCount++;
                    final String state = resource.getState();
                    if(!stoppedStateStr.equals(state)){
                        FailureDetector failureDetector = baseContext.getFailureDetector();
                        //We check use the failure detector before declaring a given services has failed.
                        if(!failureDetector.isHealthy(resource.getManagerAgentURL()) || timeSinceLastUpdate > managerEpoch5){
                            dLogger.info("["+managerContext + "]Decide resource ", name.toString() , " has crashed (no heartbeat for "+timeSinceLastUpdate+"ms)");
                            resource.setState(SystemState.CrashedState);
                            managerContext.getRuleEngine().objectChanged(resource);
                            resourcesToRemove.add(resource.getQName());
                        }else{
                            long missedTimeSteps = timeSinceLastUpdate/ (baseContext.getManagerEpochSeconds()*1000);
                            log.warn("havent heard from the resource "+ resource.getName() 
                                    + " for "+ missedTimeSteps +" timesteps, but failure detector claim it is up ("+currentTime 
                                    + " - " + resource.getLastUpdated() + ")/"+ baseContext.getManagerEpochSeconds() + "s");
                        }
                    }
                }
            }
            StringBuffer msg = new StringBuffer();
            msg.append("Manager (").append(id).append("/").append(managerContext.hasCoordinator()).append("/")
                .append(managerContext.getResourceModel().getResources().size()).append("/").append(failedResourceCount).append(") ")
                .append(new Date().toString()).append("("+JavaInstrumentation.getMemoryUsage()).append(")")
                .append(Utils.printSucessorList(managerContext.getSucessorList())).append(" ").append(managerContext);
            //dLogger.info(msg.toString());
            log.debug(msg.toString());

            // Remove the resources which are down for some time from the system
            if(resourcesToRemove.size() > 0){
                for(QName name:resourcesToRemove){
                    resourceModel.removeResource(name);
                }
                resourcesToRemove.clear();
            }
            
            //If there is no coordinator, we try to find one with anycast
            if(!managerContext.hasCoordinator()){
                 if (managerContext.getOldCoordinator() != null && Utils.selectWithProbability(0.5f)) {
                     /**
                      * Ok this is not a initialization, First we will search for a Coordinator
                      */
                    GetCoordinatorResponse response = ManagementRingTools.findCoordinator(managerContext,2);
                    if(response != null ){
                        ManagerAddress coordinatorAddress = response.getAddress();
                         if(coordinatorAddress != null && !Utils.equals(managerContext.getOldCoordinator(), coordinatorAddress)){
                             if(managerContext.getMyRank() <= coordinatorAddress.getRank()){
                                 managerContext.setCoordinatorAddress(coordinatorAddress,
                                         response.getCoordinatorTimestamp());
                                 dLogger.info("["+managerContext + "]Found a coordinator " + coordinatorAddress.getSoapAddress());
                             }else{
                                 log.warn("["+managerContext + "]Found a coordinator whose rank is higher than my myrank" + managerContext.getMyRank() + " > "+
                                         coordinatorAddress.getRank());
                                 //TODO may be we should start a election here
                             }
                         }
                     }
                 }
            }
                 
            /**
             * If we have a Coordinator, we send heart beat messages
             */
            if (managerContext.hasCoordinator()) {
                point12 = System.nanoTime();
                coodrinatorAddress = managerContext.getCoordinatorAddress();
                coordinatorTimestamp = managerContext.getCoordinatorTimeStamp();
                try {
                    ManagementRingTools.sendHeartbeatsAndPorcessPiggyback(managerContext, coodrinatorAddress.getSoapAddress());
                    point13 = System.nanoTime();
                } catch (HasthiException e) {
                    boolean errorHandled = false;
                    /**
                     * Following code handles the error if it is due to sending heartbeats to a wrong coordinator
                     * (May be coordinator just stepped down)
                     */
                    if(e.getFaultCode().equals(FaultCode.UnsupportedMessage)){
                        if(e.getMessage().contains(FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR)){
                            String currentCoordinator = e.getProperty(FAULT_DETAIL_CURRENT_COORDINATOR);
                            if(currentCoordinator != null){
                                if(!FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY.equals(currentCoordinator)){
                                    //We join the new coordinator
                                    try {
                                        ManagementRingTools.joinUsingHeartbeat(managerContext, currentCoordinator);
                                    } catch (Throwable e1) {
                                        log.fatal("error handling error while connecting to the coordinator, this can be a bug",e1);
                                    }
                                }else{
                                    //This case the coordinator is in nominated mode and asked us to retry. So we do nothing
                                    // and we will retry next epoch
                                    errorHandled = true;    
                                }
                            }
                        }else{
                            log.fatal("["+managerContext + "]Possible error, unsupported message to manager heartbeat, but not "+
                                    FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR, e);
                        }
                    }else if(!e.getFaultCode().equals(FaultCode.ServiceNotReachable)
                            && !e.getFaultCode().equals(FaultCode.ErrorAtClientWhileWsCall)){
                        dLogger.warrning("["+managerContext + "]Falied to contact Coordinator("
                                + managerContext.getCoordinatorAddress().getSoapAddress() + ")", e);
                    }
                    if(!errorHandled){
                        handleCoodinatorFaluire(e);
                    }
                }
            }else{
                /**
                 * We can not found a coordinator, if there is no
                 * election, or election has timed out we will initiate
                 * a new Coordinator election with some small
                 * probability
                 */
                if(managerContext.getManagerState().equals(MngState.NoCoordWithoutElection)){
                    initateElectionProbabilistically();    
                }
            }
            
            //Utils.verifyListIsSorted(managerContext.getSucessorList());
            /**
             * fire the local rules
             */
            long point2 = System.nanoTime();
            managerContext.getRuleEngine().fire();
            long point3 = System.nanoTime();
            loc.addAReading(point3 - point1);
            roc.addAReading(point3 - point2);
            if(point12 != -1 && point13 != -1){
                moc.addAReading(point13 - point12);
            }
        } catch (Throwable e) {
            dLogger.fatal("["+managerContext + "]Error at Managment Deamon", e);
        }
    }

 

    private void initateElectionProbabilistically() throws RemoteProcessorException, HasthiException {
        
        if (Utils.selectWithProbability(1.0f / managerContext.getManagementRingSize())) {
            boolean coordinatorInvitationSucessful = ManagementRingTools.inviteSucessor(managerContext,
                    null,managerContext.getCoordinatorAddress(),managerContext.getCoordinatorTimeStamp());
            /**
             * Start a new election
             */
            if(!coordinatorInvitationSucessful){
                //this code starts a election if it is not alredy started
                ElectionController electionController = managerContext.getElectionController();
                electionController.startElection(coodrinatorAddress,coordinatorTimestamp,false);
            }
            
        }
    }
    
 

    private void handleCoodinatorFaluire(HasthiException e) throws RemoteProcessorException,
            HasthiException {
        try {
            
            //TODO if coordinator is myself, we just exits. The thread is shutdown
            if(!Utils.equals(managerContext.getCoordinatorAddress(),coodrinatorAddress)){
                // this means we received a new coordinator while sending heartbeats. So we can ignore the lost error
                return;
            }
            log.info("Enter: handleCoodinatorFaluire");
            ManagerPingResponse coordinatorState = VerificationTools.veirfyManagerISAvalible(managerContext,coodrinatorAddress);
            if (coordinatorState == null) {
                managerContext.setCoordinatorDown();
                // TODO Check with randomly picked manager does the
                // coordinator down
                boolean coordinatorInvitationSucessful = ManagementRingTools.inviteSucessor(managerContext,null,coodrinatorAddress,coordinatorTimestamp);
                if (!coordinatorInvitationSucessful) {
                    //  Means all sucessors are down,we call for a nomination and start a election. We do a random wait before doing this.
                    //And check coordinator is alredy found.
                    int sleeptime = Math.min(managerContext.getMyRank()*20 + Utils.randomInt(10000),15000);
                    if(sleeptime > 50){
                        Thread.sleep(sleeptime);
                        log.info(managerContext + " sleep for  "+ sleeptime + "ms");
                        if(managerContext.hasCoordinator()){
                            log.info("["+managerContext + "]Coordinator has found");
                        }
                    }
                    //TODO may be we need to do this with random probabaility to avoid everyone starting a election
                    ElectionController electionController = managerContext.getElectionController();
                    boolean nominationStarted = electionController.startElection(coodrinatorAddress,coordinatorTimestamp,false);
                    if(!nominationStarted){
                        log.fatal("["+managerContext + "]Nomination must start without blocking, Possible error at election controller ");
                    }
                    System.out.println("["+managerContext + "]Start a Nomination at "+ managerContext );
                }else{
                    System.out.println("["+managerContext + "]End: invite a sucessor at "+ managerContext );
                }
            }else{
                log.info("["+managerContext + "]Manager Heartbeat failed with "+e.getMessage() + " verification says he is up");
            }
            log.info("Done: handleCoodinatorFaluire");
            if(Utils.equals(coodrinatorAddress, managerContext.getMyAddress())){
                log.fatal("Coordinator down, but his manager is up");
//                this.cancel();
                return;
            }
        }  catch (InterruptedException e1) {
            throw new HasthiException(e,FaultCode.LocalError);
        }

    }

    


}
